/**
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.zookeeper.recipes.lock;

import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import static org.apache.zookeeper.CreateMode.EPHEMERAL_SEQUENTIAL;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import org.apache.zookeeper.CreateMode;


import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.*; 

/**
 * A <a href="package.html">protocol to implement an exclusive
 *  write lock or to elect a leader</a>. <p/> You invoke {@link #lock()} to 
 *  start the process of grabbing the lock; you may get the lock then or it may be 
 *  some time later. <p/> You can register a listener so that you are invoked 
 *  when you get the lock; otherwise you can ask if you have the lock
 *  by calling {@link #isOwner()}
 *
 */
public class WriteLock extends ProtocolSupport {
    private static final Logger LOG = Logger.getLogger(WriteLock.class);

    private final String dir;
    private String id;
    private ZNodeName idName;
    private String ownerId;
    private String lastChildId;
    private byte[] data = {0x12, 0x34};
    private LockListener callback;
    private LockZooKeeperOperation zop;
    private final String meta_no = "/" + "meta" +  "x-" + "00000000000000000" + "-" + "0000000000";//falta só colocar o diretorio na frente
    
    /**
     * zookeeper contructor for writelock
     * @param zookeeper zookeeper client instance
     * @param dir the parent path you want to use for locking
     * @param acls the acls that you want to use for all the paths, 
     * if null world read/write is used.
     */
    public WriteLock(ZooKeeper zookeeper, String dir, List<ACL> acl) {
        super(zookeeper);
        this.dir = dir;
        if (acl != null) {
            setAcl(acl);
        }
        this.zop = new LockZooKeeperOperation();
    }
    
    /**
     * zookeeper contructor for writelock with callback
     * @param zookeeper the zookeeper client instance
     * @param dir the parent path you want to use for locking
     * @param acl the acls that you want to use for all the paths
     * @param callback the call back instance
     */
    public WriteLock(ZooKeeper zookeeper, String dir, List<ACL> acl, 
            LockListener callback) {
        this(zookeeper, dir, acl);
        this.callback = callback;
    }

    /**
     * return the current locklistener
     * @return the locklistener
     */
    public LockListener getLockListener() {
        return this.callback;
    }
    
    /**
     * register a different call back listener
     * @param callback the call back instance
     */
    public void setLockListener(LockListener callback) {
        this.callback = callback;
    }

    /**
     * Removes the lock or associated znode if 
     * you no longer require the lock. this also 
     * removes your request in the queue for locking
     * in case you do not already hold the lock.
     * @throws RuntimeException throws a runtime exception
     * if it cannot connect to zookeeper.
     */
    public synchronized void unlock() throws RuntimeException {
        int i,maior_sequencia = 0,aux = 0;
        List<String> lista = new ArrayList();
        byte[] seq;
        if (!isClosed() && id != null) {
            // we don't need to retry this operation in the case of failure
            // as ZK will remove ephemeral files and we don't wanna hang
            // this process when closing if we cannot reconnect to ZK
            try {
                 
                //FASE 6 DO WRITE_LOCK, AO FAZER UNLOCK MODIFICA O METADADO COM O NUMERO DE SEQUENCIA DO MAIOR LEITOR
                if(id.substring(id.length() - 35,id.length() -30).equals("write")){
                    List<String> names = zookeeper.getChildren(dir, false);//obtendo lista de nomes
                    for (String name : names) {
                        lista.add(name.substring(name.length() - 10, name.length())+ name.substring(0, 5));//concatena numero de sequencia com o tipo, write ou readx
                    }
                    Collections.sort(lista);
                    i = lista.size()-1;
                    while(i>0){
                        if(lista.get(i).substring(lista.get(i).length() - "write".length(),lista.get(i).length()).equals("readx")){
                            aux = Integer.parseInt(lista.get(i).substring(0,lista.get(i).length() - "write".length()));
                            if(i == lista.size()-1)
                                maior_sequencia = aux;
                            if(aux > maior_sequencia)
                                maior_sequencia = aux;
                        }
                        i--;
                    }
                    
                    seq = String.valueOf(maior_sequencia).getBytes();
                    zookeeper.setData(dir + meta_no, seq, -1);//muda o metadado com o novo numero de sequencia
                    //impressao opcional
                    System.out.println("\nNovo valor no meta_no: " + maior_sequencia);
                }
                //FIM DA MODIFICACAO DA FASE 6
                ZooKeeperOperation zopdel = new ZooKeeperOperation() {
                    public boolean execute() throws KeeperException,
                        InterruptedException {
                        zookeeper.delete(id, -1);   
                        return Boolean.TRUE;
                    }
                };
                zopdel.execute();
            } catch (InterruptedException e) {
                LOG.warn("Caught: " + e, e);
                //set that we have been interrupted.
               Thread.currentThread().interrupt();
            } catch (KeeperException.NoNodeException e) {
                // do nothing
            } catch (KeeperException e) {
                LOG.warn("Caught: " + e, e);
                throw (RuntimeException) new RuntimeException(e.getMessage()).
                    initCause(e);
            }
            finally {
                if (callback != null) {
                    callback.lockReleased();
                }
                id = null;
            }
        }
    }
    
    /** 
     * the watcher called on  
     * getting watch while watching 
     * my predecessor
     */
    private class LockWatcher implements Watcher {
        public void process(WatchedEvent event) {
            // lets either become the leader or watch the new/updated node
            LOG.debug("Watcher fired on path: " + event.getPath() + " state: " + 
                    event.getState() + " type " + event.getType());
            try {
                if(id != null)//MUDANÇA DO GRUPO 1 PARA PERMITIR DESISTENCIA DE LOCK
                    lock();
            } catch (Exception e) {
                LOG.warn("Failed to acquire lock: " + e, e);
            }
        }
    }
    
    /**
     * a zoookeeper operation that is mainly responsible
     * for all the magic required for locking.
     */
    private  class LockZooKeeperOperation implements ZooKeeperOperation {
        public String type;
        public String diretorio_meta_no = dir + meta_no;
        private byte[] seq;
        /** find if we have been created earler if not create our node
         * 
         * @param prefix the prefix node
         * @param zookeeper teh zookeeper client
         * @param dir the dir paretn
         * @throws KeeperException
         * @throws InterruptedException
         */
        private void findPrefixInChildren(String prefix, ZooKeeper zookeeper, String dir) 
            throws KeeperException, InterruptedException {
            List<String> names = zookeeper.getChildren(dir, false);
            for (String name : names) {
                if (name.startsWith(prefix)) {
                    id = name;
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("Found id created last time: " + id);
                    }
                    break;
                }
            }
            if (id == null) {
                if(zop.type.equals("meta")){//CRIACAO DO NOH PARA METADADOS
                    id = zookeeper.create(diretorio_meta_no, data, 
                    getAcl(), CreateMode.PERSISTENT);
                    seq = String.valueOf(0).getBytes();//inicialmente tem o valor 0
                    zookeeper.setData(id, seq, -1);//inicia o numero de sequecia do meta-dados com 0
                }else{
                    id = zookeeper.create(dir + "/" + prefix, data, 
								                getAcl(), EPHEMERAL_SEQUENTIAL);
                }
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Created id: " + id);
                }
            }
        }
        
        /**
         * the command that is run and retried for actually 
         * obtaining the lock
         * @return if the command was successful or not
         */

				
        public boolean execute() throws KeeperException, InterruptedException {
            List<String> lista = new ArrayList();
            int posicao=0;
            int i=0,j,sequencia_int;
            boolean read_pode_pegar_lock = true;//caso não exista write na fila, pega lock, por isso deixo true
            boolean tem_gente_na_frente = false;
            String nome_a_frente="ninguem";
            String meu_tipo;//se sou um write, read lock ou um inicia lock
            do {
                /*se id == null cria um novo znode (etapa 1 do read e write lock)*/
                if (id == null) {
                    long sessionId = zookeeper.getSessionId();
                    String prefix;
                    if(zop.type.equals("read")){
                        prefix = "read" + "x-" + sessionId +"-";
                    }else if(zop.type.equals("write")){
                        prefix = "write" + "x-" + sessionId + "-";
                    }else{
                        prefix = "meta" +  "x-" + "00000000000000000" + "-" + "0000000000";
                    }

                    // lets try look up the current ID if we failed 
                    // in the middle of creating the znode
                    if(zop.type.equals("meta")){
                        if(zookeeper.exists(diretorio_meta_no,false) != null)
                            return false;
                    }
                    findPrefixInChildren(prefix, zookeeper, dir);
                    if(zop.type.equals("meta")){
                        return true;//metadados criado com sucesso, posso sair
                    }else{
                        idName = new ZNodeName(id);
                    }
                }

                if (id != null) {
                    List<String> names = zookeeper.getChildren(dir, false);
                    if (names.isEmpty()) {
                        LOG.warn("No children in: " + dir + " when we've just " +
                        "created one! Lets recreate it...");
                        // lets force the recreation of the id
                        id = null;
                    } else {
                        /*nó de metadados ja criado*/
                        if(zop.type.equals("meta")){
                            return false;
                        }
                        // lets sort them explicitly (though they do seem to come back in order ususally :)
                        SortedSet<ZNodeName> sortedNames = new TreeSet<ZNodeName>();
                        for (String name : names) {
                            lista.add(name.substring(name.length() - 10, name.length())+ name.substring(0, 5));//concatena numero de sequencia com o tipo, write ou readx
                            sortedNames.add(new ZNodeName(dir + "/" + name));
                        }
                        Collections.sort(lista);

                        /*descubro minha posição na lista*/
                        for(String numero_de_sequencia : lista){
                            //ex. if(015105150 == 015105150), descubro minha posicao na lista por mesmo numero de sequencia
                            if(Integer.parseInt(numero_de_sequencia.substring(0,numero_de_sequencia.length() - "write".length())) == Integer.parseInt(id.substring(id.length() - 10, id.length()))){
                                System.out.println("\nEu sou o: " + numero_de_sequencia);
                                posicao = i;
                                break;
                            }
                            i++;
                        }
                        j = posicao;

                        //aqui vejo se sou um write ou read
                        meu_tipo = lista.get(j).substring(lista.get(j).length()-"write".length(),lista.get(j).length());

                        //CASO WRITE_LOCK(parte da etapa 2, vê quem está na frente para depois colocar um watcher)
                        if(meu_tipo.equals("write") && j > 0){
                            //procura o lock logo na frente
                            for (String name : names) {
                                //minha posicao eh j, o que esta na minha frente eh j-1
                                if(Integer.parseInt(lista.get(j-1).substring(0,lista.get(j-1).length() - "write".length())) == Integer.parseInt(name.substring(name.length() - 10, name.length()))){
                                    nome_a_frente = name;
                                    //caso seja o meta nó ignora
                                    //System.out.println("\nnome = " + nome_a_frente + "\nmeta = " + diretorio_meta_no);
                                    if((dir + "/" +nome_a_frente).equals(diretorio_meta_no)){
                                        tem_gente_na_frente = false;
                                        nome_a_frente = "ninguem";
                                        continue;
                                    }
                                    tem_gente_na_frente = true;
                                    break;
                                }								
                            }

                            //etapa 2 do write (coloca um watcher em quem estiver na frente)
                            if (tem_gente_na_frente) {
                                lastChildId = dir + "/" +nome_a_frente; // formato: /diretorio/writex-85566826469523459-0000000460
                                if (LOG.isDebugEnabled()) {
                                    LOG.debug("watching less than me node: " + lastChildId);
                                }
                                Stat stat = zookeeper.exists(lastChildId, new LockWatcher());
                                if (stat != null) {
                                    return Boolean.FALSE;
                                } else {
                                    LOG.warn("Could not find the" +
                                		" stats for less than me: " + lastChildId);
                                }

                            } else {
                                //etapa 3 do write, olha o metadados antes de pegar o lock
                                seq = zookeeper.getData(diretorio_meta_no, false, zookeeper.exists(diretorio_meta_no,false)); //le o numero de sequencia do metadados
                                sequencia_int = Integer.parseInt(new String(seq));
                                i=lista.size() - 1;
                                while(i>0){
                                    //se for um read
                                    if(lista.get(i).substring(lista.get(i).length()-"write".length(),lista.get(i).length()).equals("readx")){
                                        //e se o numero de sequencia for menor/igual que o do metadados
                                        if(Integer.parseInt(lista.get(i).substring(0,lista.get(i).length() - "write".length())) <= sequencia_int){
                                            //para achar o diretorio do read em que deve colocar um watcher
                                            for (String name : names) {
                                                if(Integer.parseInt(lista.get(i).substring(0,lista.get(i).length() - "write".length())) == Integer.parseInt(name.substring(name.length() - 10, name.length()))){
                                                    nome_a_frente = name;
                                                    tem_gente_na_frente = true;
                                                    break;
                                                }
                                            }
                                        }
                                    }
                                    i--;
                                }

                                //etapa 4
                                if(tem_gente_na_frente == true){// ou seja, existe um read com numero de sequencia menor/igual que o metadados
                                    lastChildId = dir + "/" +nome_a_frente;
                                    //coloco um watcher neste read...
                                    if (LOG.isDebugEnabled()) {
                                        LOG.debug("watching less than me node: " + lastChildId);
                                    }
                                    Stat stat = zookeeper.exists(lastChildId, new LockWatcher());
                                    if (stat != null) {
                                        return Boolean.FALSE;
                                    } else {
                                        LOG.warn("Could not find the" +
                                            " stats for less than me: " + lastChildId);
                                    }
                                    //etapa 5
                                }else{//Como não existe um leitor com num. de seq. menor que o meta dados, pega lock
                                    if (callback != null) {
                                        callback.lockAcquired();
                                    }
                                    //coloco o lock como dono do lock
                                    ownerId = id;
                                    return Boolean.TRUE;
                                }
                            }

                            //fim do caso de um write lock, agora trato um readlock


                        //...........................................................

                            //CASO READ_LOCK, etapa 2, comparo  o numero de sequencia de readlock com o metadados
                        }else if (meu_tipo.equals("readx")){
                            seq = zookeeper.getData(diretorio_meta_no, false, zookeeper.exists(diretorio_meta_no,false)); //le o numero de sequencia do metadados
                            sequencia_int = Integer.parseInt(new String(seq));
                            System.out.println("\nNumero de sequencia no metadados: " + sequencia_int);//impressao opcional
                            if(Integer.parseInt(lista.get(j).substring(0,lista.get(j).length() - "write".length())) <= sequencia_int){
                                read_pode_pegar_lock = true;
                                //etapa 3, se j = 1 significa que não tem ninguem na frente (só o no de metadados, que não conta)
                            }else if(j == 1){
                                read_pode_pegar_lock = true;
                                //etapa 5
                            }else{
                                //procura se tem um writelock na frente (do mais perto de pegar o lock até a posiçao do readlock)
                                i = 1;//primeiro na lista
                                while(i < j){
                                    if(lista.get(i).substring(lista.get(i).length()-"write".length(),lista.get(i).length()).equals("write")){
                                        //tem write na frente...
                                        //acha a posicao do write
                                        System.out.println("\nwrite= " + lista.get(i)+ "\ni = " +i);
                                        for (String name : names) {
                                            if(Integer.parseInt(lista.get(i).substring(0,lista.get(i).length() - "write".length())) == Integer.parseInt(name.substring(name.length() - 10, name.length()))){
                                                nome_a_frente = name;
                                                read_pode_pegar_lock = false;
                                                i = j;
                                                break;
                                            }//if
                                        }//for
                                    }//if
                                    i++;
                                }//while

                                //parte 6, verifico novamente o metadados
                                seq = zookeeper.getData(diretorio_meta_no, false, zookeeper.exists(diretorio_meta_no,false)); //le o numero de sequencia do metadados
                                sequencia_int = Integer.parseInt(new String(seq));
                                if(Integer.parseInt(lista.get(j).substring(0,lista.get(j).length() - "write".length())) <= sequencia_int)
                                    read_pode_pegar_lock = true;
                            }//else
                            //pega o lock ou coloca um watcher
                            if(read_pode_pegar_lock){
                                if (callback != null) {
                                    callback.lockAcquired();
                                }
                                //coloco o lock como dono do lock
                                ownerId = id;
	                            return Boolean.TRUE;
                            }else{
                                lastChildId = dir + "/" +nome_a_frente; // formato: /diretorio/writex-85566826469523459-0000000460
                                System.out.println("\nEstou esperando: " +nome_a_frente);// apenas para ilustrar
                                if (LOG.isDebugEnabled()) {
                                    LOG.debug("watching less than me node: " + lastChildId);
                                }
                                Stat stat = zookeeper.exists(lastChildId, new LockWatcher());
                                if (stat != null) {
                                    return Boolean.FALSE;
                                } else {
                                LOG.warn("Could not find the" +
                                		" stats for less than me: " + lastChildId);
                                }
                            }										
                        }//encerra tratamento do readlock

              	    }
                }
            }
            while (id == null);
            return Boolean.FALSE;
        }
    };

    /**
     * Attempts to acquire the exclusive write lock returning whether or not it was
     * acquired. Note that the exclusive lock may be acquired some time later after
     * this method has been invoked due to the current lock owner going away.
     */
    public synchronized boolean lock() throws KeeperException, InterruptedException {
        if (isClosed()) {
            return false;
        }
        ensurePathExists(dir);
        zop.type = "write";
        return (Boolean) retryOperation(zop);
    }
    /**
     * Attempts to acquire the exclusive read lock returning whether or not it was
     * acquired. Note that the exclusive lock may be acquired some time later after
     * this method has been invoked due to the current lock owner going away.
     */
    public synchronized boolean rlock() throws KeeperException, InterruptedException {
        if (isClosed()) {
            return false;
        }
        ensurePathExists(dir);
        zop.type = "read";
        return (Boolean) retryOperation(zop);
    }

	//this one is resposible to iniciate the write/lock algorithm
	public synchronized boolean inicia() throws KeeperException, InterruptedException {
        if (isClosed()) {
            return false;
        }
        ensurePathExists(dir);
        zop.type = "meta";
        return (Boolean) retryOperation(zop);
    }


    /**
     * return the parent dir for lock
     * @return the parent dir used for locks.
     */
    public String getDir() {
        return dir;
    }

    /**
     * Returns true if this node is the owner of the
     *  lock (or the leader)
     */
    public boolean isOwner() {
        if(id.equals(ownerId)){
            return true;  
        }else{
            return false;  
        }  
    }

    /**
     * return the id for this lock
     * @return the id for this lock
     */
    public String getId() {
       return this.id;
    }
}

